package org.budo.warehouse.logic.consumer.async;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import org.budo.dubbo.protocol.async.repository.AbstractAsyncRepository;
import org.budo.dubbo.protocol.async.repository.BudoAsyncInvocation;
import org.budo.graph.annotation.SpringGraph;
import org.budo.support.lang.util.MapUtil;
import org.budo.support.spring.util.SpringUtil;
import org.budo.warehouse.logic.api.DataEntry;
import org.budo.warehouse.logic.api.DataMessage;
import org.budo.warehouse.logic.bean.LogicDynamicBeanProvider;
import org.budo.warehouse.logic.util.PipelineUtil;
import org.budo.warehouse.service.api.IDataNodeService;
import org.budo.warehouse.service.entity.DataNode;
import org.budo.warehouse.service.entity.Pipeline;
import org.springframework.stereotype.Component;

/**
 * 旧版本消息发送
 * 
 * @author lmw
 */
@Component
public class V1AsyncMessageSender {
    @SpringGraph
    public void send_message_v1(Pipeline pipeline, DataMessage dataMessage) {
        IDataNodeService dataNodeService = SpringUtil.getBean(IDataNodeService.class);
        DataNode targetDataNode = dataNodeService.findByIdCached(pipeline.getTargetDataNodeId());

        LogicDynamicBeanProvider logicDynamicBeanProvider = SpringUtil.getBean(LogicDynamicBeanProvider.class);
        String beanName = logicDynamicBeanProvider.asyncRepositoryName(targetDataNode);

        List<Map<String, Object>> v1Message = this.to_v1_message(dataMessage);
        String destinationName = PipelineUtil.destinationName(pipeline);

        AbstractAsyncRepository abstractAsyncRepository = (AbstractAsyncRepository) SpringUtil.getBean(beanName);

        BudoAsyncInvocation budoAsyncInvocation = new BudoAsyncInvocation().setMethodName("pushDataToWarehouse") //
                .setParameterTypes(new Class<?>[] { List.class }) //
                .setArguments(new Object[] { v1Message }) //
                .setCreatedAt(System.currentTimeMillis()) // 消息创建时间
                .setId(v1Message.hashCode());

        abstractAsyncRepository.sendMessage(destinationName, budoAsyncInvocation);
    }

    private List<Map<String, Object>> to_v1_message(DataMessage dataMessage) {
        List<Map<String, Object>> mapList = new ArrayList<Map<String, Object>>();
        List<DataEntry> dataEntries = dataMessage.getDataEntries();
        for (DataEntry dataEntry : dataEntries) {
            Integer rowCount = dataEntry.getRowCount();
            for (int rowIndex = 0; rowIndex < rowCount; rowIndex++) {
                Map<String, Object> rowMap = this.rowMap(dataEntry, rowIndex);

                mapList.add(rowMap);
            }
        }

        return mapList;
    }

    private Map<String, Object> rowMap(DataEntry dataEntry, int rowIndex) {
        String eventType = dataEntry.getEventType(); // 标记每一行数据的操作类型，方便MQ接收端判断类型
        Map<String, Object> rowMap = MapUtil.stringObjectMap("$eventType", eventType);

        Integer columnCount = dataEntry.getColumnCount(rowIndex);
        for (int columnIndex = 0; columnIndex < columnCount; columnIndex++) {

            String columnName = dataEntry.getColumnName(rowIndex, columnIndex);

            String columnValue;
            if ("delete".equalsIgnoreCase(eventType)) {
                columnValue = dataEntry.getColumnValueBefore(rowIndex, columnIndex);
            } else {
                columnValue = dataEntry.getColumnValueAfter(rowIndex, columnIndex);
            }

            rowMap.put(columnName, null == columnValue ? "" : columnValue); // for兼容
        }
        return rowMap;
    }
}